package com.vteba.shards.table.service;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.vteba.shards.table.bean.JsonBean;
import com.vteba.shards.table.model.ShardsTable;
import com.vteba.tx.jdbc.mybatis.cache.ShardsTableCacheService;
import com.vteba.utils.charstr.Char;
import com.vteba.utils.charstr.StringUtils;
import com.vteba.utils.common.PropUtils;
import com.vteba.utils.json.FastJsonUtils;

/**
 * 分区表执行任务
 * @author yinlei
 * @date 2014年12月1日 下午3:30:42
 */
public class ShardsTableTask implements Runnable {
	private static final Logger LOGGER = LogManager.getLogger(ShardsTableTask.class);
	
	private ZkClient zkClient;
	private ShardsTableCacheService shardsTableCacheService;
	
	public ShardsTableTask(ZkClient zkClient, ShardsTableCacheService shardsTableCacheService) {
		super();
		this.zkClient = zkClient;
		this.shardsTableCacheService = shardsTableCacheService;
	}

	@Override
	public void run() {
		String shardsTablePath = PropUtils.get("zookeeper.shards.table.path");
		if (StringUtils.isEmpty(shardsTablePath)) {
			LOGGER.warn("cannot find item[zookeeper.shards.table.path] in config.properties.");
			return;
		}
		zkClient.subscribeDataChanges(shardsTablePath, new IZkDataListener() {
			
			@Override
			public void handleDataDeleted(String dataPath) throws Exception {
				LOGGER.info("delete path [{}].", dataPath);
			}
			
			@Override
			public void handleDataChange(String dataPath, Object data) throws Exception {
				String json = "";
				if (data instanceof String) {
					json = (String) data;
				} else if (data instanceof byte[]) {
					json = new String((byte[]) data, Char.UTF8);
				} else {
					LOGGER.warn("received data is not String or byte[] format, will discard.");
				}
				if (LOGGER.isInfoEnabled()) {
					LOGGER.info("received data is[{}]", json);
				}
				JsonBean bean = FastJsonUtils.fromJson(json, JsonBean.class);
				ShardsTable shardsTable = bean.getShardsTable();
				String tableName = shardsTable.getTableName();
				int code = bean.getCode();
				if (code == 1) { // 新建
					if (LOGGER.isInfoEnabled()) {
						LOGGER.info("will add table[{}].", tableName);
					}
					shardsTableCacheService.put(tableName, shardsTable);
				} else if (code == -1) { // 删除
					if (LOGGER.isInfoEnabled()) {
						LOGGER.info("will remove table[{}].", tableName);
					}
					shardsTableCacheService.remove(tableName);
					
				} else {
					LOGGER.warn("Unknown JsonBean's code[{}].", code);
				}
			}
		});
	}

	public void setZkClient(ZkClient zkClient) {
		this.zkClient = zkClient;
	}

	public void setShardsTableCacheService(ShardsTableCacheService shardsTableCacheService) {
		this.shardsTableCacheService = shardsTableCacheService;
	}

}
